Add metrics to Kafka Source #3118

kkondaka · 2023-08-05T16:12:55Z

Description

Add metrics to Kafka.

Each thread maintains a local copy of counters which are put in a topic level common data structure every 60 seconds.
These metrics are aggregated across different threads of a topic, every 60 seconds and are exported to PluginMetrics

Resolves #3112

Issues Resolved

#3112

Check List

New functionality includes testing.
New functionality has been documented.
- New functionality has javadoc added
[ X] Commits are signed with a real name per the DCO

By submitting this pull request, I confirm that my contribution is made under the terms of the Apache 2.0 license.
For more information on following Developer Certificate of Origin and signing off your commits, please check here.

Signed-off-by: Krishna Kondaka <[email protected]>

hshardeesi · 2023-08-08T05:42:13Z

...a-plugins/src/main/java/org/opensearch/dataprepper/plugins/kafka/util/KafkaTopicMetrics.java

+            cmetrics = consumerMetricsMap.get(consumer);
+        }
+        if (cmetrics == null) {
+            return;


should these be assertions? or at least an error message. unless there is a legitimate case

This should not happen. So assertion would be better but I don't see any assertions in the code.

hshardeesi · 2023-08-08T05:44:03Z

...a-plugins/src/main/java/org/opensearch/dataprepper/plugins/kafka/util/KafkaTopicMetrics.java

+            Metric value = entry.getValue();
+            String metricName = metric.name();
+            String metricGroup = metric.group();
+            if ((metricName.contains("consumed")) ||


how about we use a map to filter-in/out metrics of interest.

Some of the names exist under two categories, so map may not be straight-forward. Let me check again.

hshardeesi · 2023-08-08T05:46:12Z

...a-plugins/src/main/java/org/opensearch/dataprepper/plugins/kafka/util/KafkaTopicMetrics.java

+        return camelCaseName;
+    }
+
+    public void update(final KafkaConsumer consumer, final String metricName, Integer metricValue) {


minor: function name can be renamed to something that differentiates functionalities of both functions.

hshardeesi · 2023-08-08T06:00:53Z

...a-plugins/src/main/java/org/opensearch/dataprepper/plugins/kafka/util/KafkaTopicMetrics.java

+        }
+        synchronized (consumerMetricsMap) {
+            long curTime = Instant.now().getEpochSecond();
+            if (curTime - updateTime > metricUpdateInterval) {


this is done in caller function as well.

Yes. It's done at two layers. One to update the global data structure and the other to do the aggregation.

hshardeesi · 2023-08-08T06:01:36Z

...a-plugins/src/main/java/org/opensearch/dataprepper/plugins/kafka/util/KafkaTopicMetrics.java

+                    });
+                });
+                aggregatedMetrics.forEach((name, value) -> {
+                    pluginMetrics.gauge("topic."+topicName+"."+getCamelCaseName(name), value);


if we are aggregating for a topic then should this function be called once per topic with list of consumers? may be call this from kafka source thread for each topic..

It is called for list of consumers, but indirectly.

hshardeesi · 2023-08-08T06:03:21Z

...a-plugins/src/main/java/org/opensearch/dataprepper/plugins/kafka/util/KafkaTopicMetrics.java

+                    cmetrics.put(metricName, value.metricValue());
+            }
+        }
+        synchronized (consumerMetricsMap) {


this aggregation can be defined in a separate function.

Signed-off-by: Krishna Kondaka <[email protected]>

…gements enabled Signed-off-by: Krishna Kondaka <[email protected]>

dlvenable · 2023-08-08T19:24:49Z

...a-plugins/src/main/java/org/opensearch/dataprepper/plugins/kafka/util/KafkaTopicMetrics.java

+                });
+            });
+            aggregatedMetrics.forEach((name, value) -> {
+                pluginMetrics.gauge("topic."+topicName+"."+getCamelCaseName(name), value);


A gauge should be setup once and then updated over time.

Discussed this offline.

dlvenable · 2023-08-08T19:27:27Z

...c/main/java/org/opensearch/dataprepper/plugins/kafka/consumer/KafkaSourceCustomConsumer.java

                     e.topicPartition().topic(), e.topicPartition().partition(), e.offset());
+            numberOfDeserializationErrors++;


Why can't we just increment a Micrometer counter?

dlvenable · 2023-08-08T19:27:32Z

...c/main/java/org/opensearch/dataprepper/plugins/kafka/consumer/KafkaSourceCustomConsumer.java

@@ -216,15 +258,34 @@ Map<TopicPartition, OffsetAndMetadata> getOffsetsToCommit() {
        return offsetsToCommit;
    }

+    public void updateMetrics() {
+        long curTime = Instant.now().getEpochSecond();
+        if (curTime - metricsUpdatedTime >= metricsUpdateInterval) {


I don't think we need to track the metrics updates. Micrometer handles this for us. If you provide a DoubleSupplier to a gauge, then Micrometer can get what it needs at that point in time.

dlvenable · 2023-08-08T23:09:22Z

...a-plugins/src/main/java/org/opensearch/dataprepper/plugins/kafka/util/KafkaTopicMetrics.java

+        camelCaseMap.put("join-rate", "joinRate");
+        camelCaseMap.put("incoming-byte-rate", "incomingByteRate");
+        camelCaseMap.put("outgoing-byte-rate", "outgoingByteRate");
+        camelCaseMap.put("assigned-partitions", "outgoingByteRate");


The value should be assignedPartitions. Maybe a copy-paste error?

Actually, I don't need this. Thanks for finding this.

Removed it.

Signed-off-by: Krishna Kondaka <[email protected]>

dlvenable · 2023-08-09T14:49:23Z

.../src/main/java/org/opensearch/dataprepper/plugins/kafka/configuration/KafkaSourceConfig.java

@@ -37,6 +37,7 @@ public boolean getInsecure() {
    }

    public static final Duration DEFAULT_ACKNOWLEDGEMENTS_TIMEOUT = Duration.ofSeconds(30);
+    public static final Duration DEFAULT_METRICS_UPDATE_INTERVAL = Duration.ofSeconds(60);


It seems this is unused and may be left over from the previous commits.

Removed it.

Signed-off-by: Krishna Kondaka <[email protected]>

…artitions assigned Signed-off-by: Krishna Kondaka <[email protected]>

dlvenable · 2023-08-09T16:31:24Z

...a-plugins/src/main/java/org/opensearch/dataprepper/plugins/kafka/util/KafkaTopicMetrics.java

+        return numberOfPositiveAcknowledgements;
+    }
+
+    public String getTopicMetricName(final String metricName) {


This can be private.

dlvenable · 2023-08-09T18:23:48Z

...a-plugins/src/main/java/org/opensearch/dataprepper/plugins/kafka/util/KafkaTopicMetrics.java

+        metricsNameMap.put("outgoing-byte-rate", "outgoingByteRate");
+        metricsNameMap.forEach((metricName, camelCaseName) -> {
+            if (!metricName.contains("-total")) {
+                pluginMetrics.gauge(getTopicMetricName(camelCaseName), metricValues, metricValues -> {


This code should be unit tested.

Also, it might make sense to invert the if-else with the gauge to clarify the behavior some. The condition is really on the metric function, not inside it.

Something like:

if(metricName.equals("records-lag-max") { pluginMetrics.gauge(getTopicMetricName(camelCaseName), metricValues, metricValues -> { double max = 0.0; for (Map.Entry<KafkaConsumer, Map<String, Double>> entry : metricValues.entrySet()) { if (entry.getValue().get(metricName) > max) { max = entry.getValue().get(metricName); } } return max; }); else if(...) pluginMetrics.gauge(...) }

dlvenable · 2023-08-09T18:28:27Z

...a-plugins/src/main/java/org/opensearch/dataprepper/plugins/kafka/util/KafkaTopicMetrics.java

+        metricsNameMap.forEach((metricName, camelCaseName) -> {
+            if (!metricName.contains("-total")) {
+                pluginMetrics.gauge(getTopicMetricName(camelCaseName), metricValues, metricValues -> {
+                    synchronized(metricValues) {


Do we need to synchronize on metricValues? Won't this collection stay the same?

Might we synchronize on the individual entries?

Yes. That's a good idea. Will try synchronize on individual entries.

dlvenable · 2023-08-09T18:30:46Z

...a-plugins/src/main/java/org/opensearch/dataprepper/plugins/kafka/util/KafkaTopicMetrics.java

+                    double newValue = (Double)value.metricValue();
+                    if (metricName.equals("records-consumed-total")) {
+                        double prevValue = consumerMetrics.get(metricName);
+                        numberOfRecordsConsumed.increment(newValue - prevValue);


Rather than tracking the previous value, do you think we could use the metric's built-in time feature?

https://kafka.apache.org/27/javadoc/org/apache/kafka/common/metrics/Measurable.html#measure-org.apache.kafka.common.metrics.MetricConfig-long-

Thus:

previousValue = ???.measure(???, lastTime);

It's not critical, but might be something to consider. Perhaps for a follow-on.

Will do this later.

Signed-off-by: Krishna Kondaka <[email protected]>

dlvenable · 2023-08-10T15:26:19Z

...gins/src/test/java/org/opensearch/dataprepper/plugins/kafka/util/KafkaTopicMetricsTests.java

+
+    @ParameterizedTest
+    @ValueSource(ints = {1, 5, 10})
+    //@ValueSource(ints = {2})


Please remove this comment. I'm fine doing this now or in a follow-on PR.

Add metrics to Kafka Source

eb19156

Signed-off-by: Krishna Kondaka <[email protected]>

kkondaka requested review from chenqi0805, engechas, graytaylor0, dinujoh, cmanning09, asifsmohammed, dlvenable and oeyh as code owners August 5, 2023 16:12

Krishna Kondaka added 3 commits August 5, 2023 16:13

Removed debug print statement

7190bee

Signed-off-by: Krishna Kondaka <[email protected]>

Fixed failing test case

6e91aac

Signed-off-by: Krishna Kondaka <[email protected]>

Added total committed metric and fixed tests

f4a727d

Signed-off-by: Krishna Kondaka <[email protected]>

hshardeesi reviewed Aug 8, 2023

View reviewed changes

Krishna Kondaka added 2 commits August 8, 2023 15:02

Addressed review comments

3272a8f

Signed-off-by: Krishna Kondaka <[email protected]>

Fixed number of committed records stat. Also fixed bug when acknowled…

b4cca7d

…gements enabled Signed-off-by: Krishna Kondaka <[email protected]>

hshardeesi approved these changes Aug 8, 2023

View reviewed changes

dlvenable requested changes Aug 8, 2023

View reviewed changes

Krishna Kondaka added 2 commits August 9, 2023 14:12

Addressed review comments. Fixed acknowledgements related bug

0b45113

Signed-off-by: Krishna Kondaka <[email protected]>

Fixed to use counters for records/bytes consumed metrics

e6c8e14

Signed-off-by: Krishna Kondaka <[email protected]>

dlvenable requested changes Aug 9, 2023

View reviewed changes

Krishna Kondaka added 2 commits August 9, 2023 16:39

Removed unused code

fec2a10

Signed-off-by: Krishna Kondaka <[email protected]>

Added a metric for keeping track of number of consumers without any p…

b9fc87f

…artitions assigned Signed-off-by: Krishna Kondaka <[email protected]>

dlvenable requested changes Aug 9, 2023

View reviewed changes

Added unit test for KafkaTopicMetrics

b2c540a

Signed-off-by: Krishna Kondaka <[email protected]>

dlvenable approved these changes Aug 10, 2023

View reviewed changes

hshardeesi approved these changes Aug 10, 2023

View reviewed changes

kkondaka merged commit 44e2eaf into opensearch-project:main Aug 10, 2023
24 checks passed

kkondaka deleted the kafka-metrics branch May 13, 2024 05:52

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Add metrics to Kafka Source #3118

Add metrics to Kafka Source #3118

kkondaka commented Aug 5, 2023

hshardeesi Aug 8, 2023

kkondaka Aug 8, 2023

hshardeesi Aug 8, 2023

kkondaka Aug 8, 2023

hshardeesi Aug 8, 2023

hshardeesi Aug 8, 2023

kkondaka Aug 8, 2023

hshardeesi Aug 8, 2023 •

edited

Loading

kkondaka Aug 8, 2023

hshardeesi Aug 8, 2023 •

edited

Loading

dlvenable Aug 8, 2023

kkondaka Aug 9, 2023

dlvenable Aug 8, 2023

dlvenable Aug 8, 2023

dlvenable Aug 8, 2023

kkondaka Aug 8, 2023

kkondaka Aug 9, 2023

dlvenable Aug 9, 2023

kkondaka Aug 9, 2023

dlvenable Aug 9, 2023

dlvenable Aug 9, 2023

dlvenable Aug 9, 2023

dlvenable Aug 9, 2023

kkondaka Aug 9, 2023

dlvenable Aug 9, 2023

kkondaka Aug 9, 2023

dlvenable Aug 10, 2023

		e.topicPartition().topic(), e.topicPartition().partition(), e.offset());
		numberOfDeserializationErrors++;

Add metrics to Kafka Source #3118

Add metrics to Kafka Source #3118

Conversation

kkondaka commented Aug 5, 2023

Description

Issues Resolved

Check List

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

hshardeesi Aug 8, 2023 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

hshardeesi Aug 8, 2023 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

hshardeesi Aug 8, 2023 •

edited

Loading

hshardeesi Aug 8, 2023 •

edited

Loading